use anyhow::Result;
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, AsyncReadExt, AsyncSeekExt, BufReader, BufWriter};
use tokio::net::TcpStream;
use std::path::Path;
use std::time::{Duration, Instant};
use std::sync::{Arc, Mutex};
use std::collections::{VecDeque, BinaryHeap};
use tokio::sync::{Semaphore, Mutex as TokioMutex};
use tracing::{info, debug};
use sha2::{Sha256, Digest};
use hex;
use std::cmp::Ordering;

use shared::types::{FtpResponse, FtpResponseCode, TransferMode};
use crate::ui::progress::TransferStatus;

/// 传输优先级
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
pub enum TransferPriority {
    Low = 1,
    Normal = 2,
    High = 3,
    Critical = 4,
}

/// 队列中的传输任务
#[derive(Debug, Clone)]
pub struct QueuedTransferTask {
    pub task: TransferTask,
    pub priority: TransferPriority,
    pub created_at: Instant,
    pub scheduled_at: Option<Instant>,
}

/// 优先级队列项（用于BinaryHeap）
#[derive(Debug, Clone)]
pub struct PriorityQueueItem {
    pub priority: TransferPriority,
    pub created_at: Instant,
    pub task_id: u64,
}

impl PartialEq for PriorityQueueItem {
    fn eq(&self, other: &Self) -> bool {
        self.priority == other.priority && self.created_at == other.created_at
    }
}

impl Eq for PriorityQueueItem {}

impl PartialOrd for PriorityQueueItem {
    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
        Some(self.cmp(other))
    }
}

impl Ord for PriorityQueueItem {
    fn cmp(&self, other: &Self) -> Ordering {
        // 优先级高的先执行，相同优先级按创建时间排序
        match self.priority.cmp(&other.priority) {
            Ordering::Equal => other.created_at.cmp(&self.created_at), // 早创建的优先
            other => other,
        }
    }
}

/// 传输任务类型
#[derive(Debug, Clone)]
pub enum TransferTask {
    Upload {
        remote_path: String,
        local_path: String,
        checksum_type: Option<ChecksumType>,
        enable_resume: bool,
    },
    Download {
        remote_path: String,
        local_path: String,
        checksum_type: Option<ChecksumType>,
        enable_resume: bool,
    },
}

/// 传输任务信息
#[derive(Debug, Clone)]
pub struct TransferTaskInfo {
    pub id: u64,
    pub task: TransferTask,
    pub priority: TransferPriority,
    pub status: TransferStatus,
    pub progress: f64, // 0.0 - 1.0
    pub bytes_transferred: u64,
    pub total_bytes: Option<u64>,
    pub speed: f64, // bytes per second
    pub error: Option<String>,
    pub result: Option<TransferResult>,
    pub created_at: Instant,
    pub started_at: Option<Instant>,
    pub completed_at: Option<Instant>,
}

/// 并发传输管理器（性能优化版本）
#[derive(Debug, Clone)]
pub struct ConcurrentTransferManager {
    max_concurrent: usize,
    semaphore: Arc<Semaphore>,
    priority_queue: Arc<TokioMutex<BinaryHeap<PriorityQueueItem>>>,
    task_map: Arc<TokioMutex<std::collections::HashMap<u64, TransferTaskInfo>>>,
    running_tasks: Arc<TokioMutex<Vec<TransferTaskInfo>>>,
    completed_tasks: Arc<TokioMutex<Vec<TransferTaskInfo>>>,
    next_task_id: Arc<Mutex<u64>>,
    queue_processor_handle: Arc<TokioMutex<Option<tokio::task::JoinHandle<()>>>>,
    // 性能优化字段
    buffer_pool: Arc<Mutex<Vec<Vec<u8>>>>, // 缓冲区池
    batch_size: usize, // 批处理大小
    prefetch_enabled: bool, // 预取启用
    compression_enabled: bool, // 压缩启用
    transfer_stats: Arc<Mutex<TransferStats>>, // 传输统计
}

/// 传输统计信息
#[derive(Debug, Clone)]
pub struct TransferStats {
    pub total_transfers: u64,
    pub successful_transfers: u64,
    pub failed_transfers: u64,
    pub total_bytes: u64,
    pub average_speed: f64,
    pub peak_speed: f64,
    pub total_duration: Duration,
}

impl Default for TransferStats {
    fn default() -> Self {
        Self {
            total_transfers: 0,
            successful_transfers: 0,
            failed_transfers: 0,
            total_bytes: 0,
            average_speed: 0.0,
            peak_speed: 0.0,
            total_duration: Duration::from_secs(0),
        }
    }
}

/// 校验和类型
#[derive(Debug, Clone, PartialEq)]
pub enum ChecksumType {
    Md5,
    Sha256,
}

/// 校验和结果
#[derive(Debug, Clone)]
pub struct ChecksumResult {
    pub checksum_type: ChecksumType,
    pub checksum: String,
}

/// 传输结果
#[derive(Debug, Clone)]
pub struct TransferResult {
    pub bytes_transferred: u64,
    pub duration: Duration,
    pub speed: f64, // bytes per second
    pub resumed: bool, // 是否从断点续传
    pub resume_offset: u64, // 续传偏移量
    pub checksum: Option<ChecksumResult>, // 校验和结果
    pub integrity_verified: bool, // 完整性验证结果
}

pub struct FtpClient {
    control_stream: Option<BufReader<TcpStream>>,
    pub session: Option<ClientSession>,
    is_connected: bool,
    is_authenticated: bool,
    concurrent_manager: Option<ConcurrentTransferManager>,
}

pub struct ClientSession {
    pub username: String,
    pub current_directory: String,
    pub transfer_mode: shared::types::TransferMode,
    pub server_features: Vec<String>,
    pub passive_port: Option<u16>,
    pub passive_ip: Option<String>,
}

impl FtpClient {
    pub fn new() -> Self {
        Self {
            control_stream: None,
            session: None,
            is_connected: false,
            is_authenticated: false,
            concurrent_manager: None,
        }
    }
    
    /// 初始化并发传输管理器
    pub fn init_concurrent_transfers(&mut self, max_concurrent: usize) {
        self.concurrent_manager = Some(ConcurrentTransferManager::new(max_concurrent));
    }
    
    /// 添加并发上传任务
    pub async fn add_concurrent_upload(&mut self, remote_path: String, local_path: String, checksum_type: Option<ChecksumType>, enable_resume: bool, priority: TransferPriority) -> Result<u64> {
        if let Some(manager) = &self.concurrent_manager {
            let task = TransferTask::Upload {
                remote_path,
                local_path,
                checksum_type,
                enable_resume,
            };
            Ok(manager.add_task(task, priority).await)
        } else {
            Err(anyhow::anyhow!("Concurrent transfer manager not initialized"))
        }
    }
    
    /// 添加并发下载任务
    pub async fn add_concurrent_download(&mut self, remote_path: String, local_path: String, checksum_type: Option<ChecksumType>, enable_resume: bool, priority: TransferPriority) -> Result<u64> {
        if let Some(manager) = &self.concurrent_manager {
            let task = TransferTask::Download {
                remote_path,
                local_path,
                checksum_type,
                enable_resume,
            };
            Ok(manager.add_task(task, priority).await)
        } else {
            Err(anyhow::anyhow!("Concurrent transfer manager not initialized"))
        }
    }
    
    /// 添加并发上传任务（默认优先级）
    pub async fn add_concurrent_upload_default(&mut self, remote_path: String, local_path: String, checksum_type: Option<ChecksumType>, enable_resume: bool) -> Result<u64> {
        self.add_concurrent_upload(remote_path, local_path, checksum_type, enable_resume, TransferPriority::Normal).await
    }
    
    /// 添加并发下载任务（默认优先级）
    pub async fn add_concurrent_download_default(&mut self, remote_path: String, local_path: String, checksum_type: Option<ChecksumType>, enable_resume: bool) -> Result<u64> {
        self.add_concurrent_download(remote_path, local_path, checksum_type, enable_resume, TransferPriority::Normal).await
    }
    
    /// 执行并发传输
    pub async fn execute_concurrent_transfers(&mut self) -> Result<()> {
        if let Some(manager) = self.concurrent_manager.take() {
            let pending_tasks = manager.get_pending_tasks().await;
            
            for task_info in pending_tasks {
                let manager_clone = manager.clone();
                let task_id = task_info.id;
                let task = task_info.task.clone();
                
                // 获取信号量许可
                let _permit = manager.semaphore.acquire().await?;
                
                // 启动传输任务
                tokio::spawn(async move {
                    match &task {
                        TransferTask::Upload { remote_path: _, local_path: _, checksum_type: _, enable_resume: _ } => {
                            // 这里需要实现实际的并发上传逻辑
                            // 由于需要访问FtpClient的方法，我们需要重构这部分
                            manager_clone.update_task_status(task_id, TransferStatus::Running).await.unwrap();
                            
                            // 模拟传输过程
                            tokio::time::sleep(Duration::from_secs(2)).await;
                            
                            // 模拟完成
                            let result = TransferResult {
                                bytes_transferred: 1024,
                                duration: Duration::from_secs(2),
                                speed: 512.0,
                                resumed: false,
                                resume_offset: 0,
                                checksum: None,
                                integrity_verified: true,
                            };
                            
                            manager_clone.set_task_result(task_id, result).await.unwrap();
                        }
                        TransferTask::Download { remote_path: _, local_path: _, checksum_type: _, enable_resume: _ } => {
                            // 这里需要实现实际的并发下载逻辑
                            manager_clone.update_task_status(task_id, TransferStatus::Running).await.unwrap();
                            
                            // 模拟传输过程
                            tokio::time::sleep(Duration::from_secs(3)).await;
                            
                            // 模拟完成
                            let result = TransferResult {
                                bytes_transferred: 2048,
                                duration: Duration::from_secs(3),
                                speed: 682.67,
                                resumed: false,
                                resume_offset: 0,
                                checksum: None,
                                integrity_verified: true,
                            };
                            
                            manager_clone.set_task_result(task_id, result).await.unwrap();
                        }
                    }
                });
            }
            
            // 将manager放回
            self.concurrent_manager = Some(manager);
            Ok(())
        } else {
            Err(anyhow::anyhow!("Concurrent transfer manager not initialized"))
        }
    }
    
    /// 获取传输任务状态
    pub async fn get_transfer_status(&self) -> Result<Vec<TransferTaskInfo>> {
        if let Some(manager) = &self.concurrent_manager {
            Ok(manager.get_all_tasks().await)
        } else {
            Err(anyhow::anyhow!("Concurrent transfer manager not initialized"))
        }
    }
    
    /// 等待所有传输完成
    pub async fn wait_for_transfers(&self) -> Result<()> {
        if let Some(manager) = &self.concurrent_manager {
            loop {
                let running_tasks = manager.get_running_tasks().await;
                if running_tasks.is_empty() {
                    break;
                }
                tokio::time::sleep(Duration::from_millis(100)).await;
            }
            Ok(())
        } else {
            Err(anyhow::anyhow!("Concurrent transfer manager not initialized"))
        }
    }
    
    /// 检查是否已连接
    pub fn is_connected(&self) -> bool {
        self.is_connected
    }
    
    /// 检查是否已认证
    pub fn is_authenticated(&self) -> bool {
        self.is_authenticated
    }
    
    pub async fn connect(&mut self, host: &str, port: u16) -> Result<()> {
        let addr = format!("{}:{}", host, port);
        let stream = TcpStream::connect(&addr).await?;
        
        self.control_stream = Some(BufReader::new(stream));
        
        // 读取欢迎消息
        let response = self.read_response().await?;
        if response.code != FtpResponseCode::ServiceReady {
            return Err(anyhow::anyhow!("Connection failed: {}", response.message));
        }
        
        self.is_connected = true;
        info!("Connected to FTP server at {}:{}", host, port);
        Ok(())
    }
    
    pub async fn login(&mut self, username: &str, password: &str) -> Result<()> {
        if !self.is_connected {
            return Err(anyhow::anyhow!("Not connected to server"));
        }
        
        // USER命令
        self.send_command(&format!("USER {}", username)).await?;
        let response = self.read_response().await?;
        
        if response.code == FtpResponseCode::UserNameOk {
            // PASS命令
            self.send_command(&format!("PASS {}", password)).await?;
            let response = self.read_response().await?;
            
            if response.code == FtpResponseCode::UserLoggedIn {
                self.session = Some(ClientSession {
                    username: username.to_string(),
                    current_directory: "/".to_string(),
                    transfer_mode: TransferMode::Passive,
                    server_features: Vec::new(),
                    passive_port: None,
                    passive_ip: None,
                });
                self.is_authenticated = true;
                info!("Successfully logged in as {}", username);
                Ok(())
            } else {
                Err(anyhow::anyhow!("Authentication failed: {}", response.message))
            }
        } else {
            Err(anyhow::anyhow!("Authentication failed: {}", response.message))
        }
    }
    
    pub async fn quit(&mut self) -> Result<()> {
        if self.is_connected {
            self.send_command("QUIT").await?;
            let _response = self.read_response().await?;
            self.is_connected = false;
            self.is_authenticated = false;
            self.session = None;
            info!("Disconnected from FTP server");
        }
        Ok(())
    }
    
    /// 获取当前工作目录
    pub async fn pwd(&mut self) -> Result<String> {
        self.ensure_authenticated()?;
        
        self.send_command("PWD").await?;
        let response = self.read_response().await?;
        
        if response.code == FtpResponseCode::PathnameCreated {
            // 解析路径，格式: "257 "/path" is current directory"
            if let Some(start) = response.message.find('"') {
                if let Some(end) = response.message[start+1..].find('"') {
                    let path = &response.message[start+1..start+1+end];
                    if let Some(session) = &mut self.session {
                        session.current_directory = path.to_string();
                    }
                    return Ok(path.to_string());
                }
            }
        }
        
        Err(anyhow::anyhow!("Failed to get current directory: {}", response.message))
    }
    
    /// 改变目录
    pub async fn cwd(&mut self, path: &str) -> Result<()> {
        self.ensure_authenticated()?;
        
        self.send_command(&format!("CWD {}", path)).await?;
        let response = self.read_response().await?;
        
        if response.code == FtpResponseCode::RequestedFileActionOk {
            if let Some(session) = &mut self.session {
                session.current_directory = path.to_string();
            }
            Ok(())
        } else {
            Err(anyhow::anyhow!("Failed to change directory: {}", response.message))
        }
    }
    
    /// 列出目录内容
    pub async fn list(&mut self, path: Option<&str>) -> Result<Vec<String>> {
        self.ensure_authenticated()?;
        
        let command = match path {
            Some(p) => format!("LIST {}", p),
            None => "LIST".to_string(),
        };
        
        self.send_command(&command).await?;
        let response = self.read_response().await?;
        
        if response.code == FtpResponseCode::DataConnectionAlreadyOpen {
            // 这里应该建立数据连接并读取目录列表
            // 简化实现，返回成功响应
            Ok(vec!["Directory listing would be here".to_string()])
        } else {
            Err(anyhow::anyhow!("Failed to list directory: {}", response.message))
        }
    }
    
    /// 设置传输模式
    pub async fn set_transfer_mode(&mut self, mode: TransferMode) -> Result<()> {
        self.ensure_authenticated()?;
        
        match mode {
            TransferMode::Passive => {
                self.send_command("PASV").await?;
                let response = self.read_response().await?;
                
                if response.code == FtpResponseCode::EnteringPassiveMode {
                    // 解析PASV响应: "227 Entering Passive Mode (h1,h2,h3,h4,p1,p2)"
                    if let Some(start) = response.message.find('(') {
                        if let Some(end) = response.message.find(')') {
                            let pasv_info = &response.message[start+1..end];
                            let parts: Vec<&str> = pasv_info.split(',').collect();
                            
                            if parts.len() == 6 {
                                let ip = format!("{}.{}.{}.{}", parts[0], parts[1], parts[2], parts[3]);
                                let port = (parts[4].parse::<u16>().unwrap_or(0) << 8) + parts[5].parse::<u16>().unwrap_or(0);
                                
                                if let Some(session) = &mut self.session {
                                    session.transfer_mode = TransferMode::Passive;
                                    session.passive_ip = Some(ip);
                                    session.passive_port = Some(port);
                                }
                                
                                info!("Passive mode set: {}:{}", 
                                    self.session.as_ref().unwrap().passive_ip.as_ref().unwrap(),
                                    self.session.as_ref().unwrap().passive_port.unwrap()
                                );
                            }
                        }
                    }
                    Ok(())
                } else {
                    Err(anyhow::anyhow!("Failed to set passive mode: {}", response.message))
                }
            }
            TransferMode::Active => {
                // 简化实现，实际需要解析PORT命令
                Err(anyhow::anyhow!("Active mode not implemented yet"))
            }
        }
    }
    
    /// 设置传输类型
    pub async fn set_transfer_type(&mut self, transfer_type: &str) -> Result<()> {
        self.ensure_authenticated()?;
        
        self.send_command(&format!("TYPE {}", transfer_type)).await?;
        let response = self.read_response().await?;
        
        if response.code == FtpResponseCode::CommandOk {
            Ok(())
        } else {
            Err(anyhow::anyhow!("Failed to set transfer type: {}", response.message))
        }
    }
    
    /// 设置断点续传偏移量
    pub async fn set_resume_offset(&mut self, offset: u64) -> Result<()> {
        self.ensure_authenticated()?;
        
        self.send_command(&format!("REST {}", offset)).await?;
        let response = self.read_response().await?;
        
        if response.code == FtpResponseCode::RequestedFileActionOk {
            info!("Resume offset set to {} bytes", offset);
            Ok(())
        } else {
            Err(anyhow::anyhow!("Failed to set resume offset: {}", response.message))
        }
    }

    /// 创建目录
    pub async fn mkdir(&mut self, path: &str) -> Result<()> {
        self.ensure_authenticated()?;
        
        self.send_command(&format!("MKD {}", path)).await?;
        let response = self.read_response().await?;
        if response.code == FtpResponseCode::PathnameCreated {
            Ok(())
        } else {
            Err(anyhow::anyhow!("Failed to create directory: {}", response.message))
        }
    }

    /// 删除目录
    pub async fn rmdir(&mut self, path: &str) -> Result<()> {
        self.ensure_authenticated()?;
        
        self.send_command(&format!("RMD {}", path)).await?;
        let response = self.read_response().await?;
        if response.code == FtpResponseCode::RequestedFileActionOk {
            Ok(())
        } else {
            Err(anyhow::anyhow!("Failed to remove directory: {}", response.message))
        }
    }

    /// 删除文件
    pub async fn delete(&mut self, filename: &str) -> Result<()> {
        self.ensure_authenticated()?;
        
        self.send_command(&format!("DELE {}", filename)).await?;
        let response = self.read_response().await?;
        if response.code == FtpResponseCode::RequestedFileActionOk {
            Ok(())
        } else {
            Err(anyhow::anyhow!("Failed to delete file: {}", response.message))
        }
    }

    /// 设置数据端口（主动模式）
    pub async fn set_data_port(&mut self, port: u16) -> Result<()> {
        self.ensure_authenticated()?;
        
        self.send_command(&format!("PORT 127,0,0,1,{},{}", port >> 8, port & 0xFF)).await?;
        let response = self.read_response().await?;
        if response.code == FtpResponseCode::CommandOk {
            Ok(())
        } else {
            Err(anyhow::anyhow!("Failed to set data port: {}", response.message))
        }
    }

    /// 获取文件大小
    pub async fn size(&mut self, filename: &str) -> Result<u64> {
        self.ensure_authenticated()?;
        
        self.send_command(&format!("SIZE {}", filename)).await?;
        let response = self.read_response().await?;
        if response.code == FtpResponseCode::FileStatusOk {
            response.message.parse::<u64>()
                .map_err(|e| anyhow::anyhow!("Failed to parse file size: {}", e))
        } else {
            Err(anyhow::anyhow!("Failed to get file size: {}", response.message))
        }
    }

    /// 获取文件修改时间
    pub async fn mdtm(&mut self, filename: &str) -> Result<String> {
        self.ensure_authenticated()?;
        
        self.send_command(&format!("MDTM {}", filename)).await?;
        let response = self.read_response().await?;
        if response.code == FtpResponseCode::FileStatusOk {
            Ok(response.message.trim().to_string())
        } else {
            Err(anyhow::anyhow!("Failed to get file modification time: {}", response.message))
        }
    }
    
    /// 检查文件是否存在并获取大小
    pub async fn get_file_size(&mut self, filename: &str) -> Result<Option<u64>> {
        self.ensure_authenticated()?;
        
        self.send_command(&format!("SIZE {}", filename)).await?;
        let response = self.read_response().await?;
        
        if response.code == FtpResponseCode::FileStatusOk {
            if let Ok(size) = response.message.trim().parse::<u64>() {
                Ok(Some(size))
            } else {
                Ok(None)
            }
        } else {
            Ok(None) // 文件不存在或无法获取大小
        }
    }
    
    /// 上传文件（支持断点续传）
    pub async fn upload_file(&mut self, local_path: &Path, remote_filename: &str) -> Result<TransferResult> {
        self.upload_file_with_resume(local_path, remote_filename, true).await
    }
    
    /// 上传文件（带断点续传选项）
    pub async fn upload_file_with_resume(&mut self, local_path: &Path, remote_filename: &str, enable_resume: bool) -> Result<TransferResult> {
        self.upload_file_with_resume_and_checksum(local_path, remote_filename, enable_resume, None).await
    }
    
    /// 上传文件（带断点续传和校验和选项）
    pub async fn upload_file_with_resume_and_checksum(&mut self, local_path: &Path, remote_filename: &str, enable_resume: bool, checksum_type: Option<ChecksumType>) -> Result<TransferResult> {
        self.ensure_authenticated()?;
        
        // 确保使用被动模式
        self.set_transfer_mode(TransferMode::Passive).await?;
        
        // 打开本地文件
        let mut file = tokio::fs::File::open(local_path).await?;
        let file_size = file.metadata().await?.len();
        
        // 检查断点续传
        let mut resume_offset = 0u64;
        let mut resumed = false;
        
        if enable_resume {
            // 检查远程文件是否存在
            if let Some(remote_size) = self.get_file_size(remote_filename).await? {
                if remote_size < file_size {
                    resume_offset = remote_size;
                    resumed = true;
                    info!("Resuming upload from offset {} bytes", resume_offset);
                    
                    // 设置REST命令
                    self.set_resume_offset(resume_offset).await?;
                    
                    // 跳过已传输的部分
                    file.seek(std::io::SeekFrom::Start(resume_offset)).await?;
                }
            }
        }
        
        // 发送STOR命令
        self.send_command(&format!("STOR {}", remote_filename)).await?;
        let response = self.read_response().await?;
        
        if response.code != FtpResponseCode::DataConnectionAlreadyOpen {
            return Err(anyhow::anyhow!("Failed to start upload: {}", response.message));
        }
        
        // 建立数据连接
        let data_stream = self.create_data_connection().await?;
        let mut data_writer = BufWriter::new(data_stream);
        
        // 传输文件
        let start_time = Instant::now();
        let mut buffer = [0u8; 8192];
        let mut total_bytes = resume_offset;
        
        loop {
            let bytes_read = file.read(&mut buffer).await?;
            if bytes_read == 0 {
                break;
            }
            
            data_writer.write_all(&buffer[..bytes_read]).await?;
            total_bytes += bytes_read as u64;
            
            // 显示进度
            let progress = (total_bytes as f64 / file_size as f64) * 100.0;
            if resumed {
                info!("Resume upload progress: {:.1}% ({}/{} bytes)", progress, total_bytes, file_size);
            } else {
                info!("Upload progress: {:.1}% ({}/{} bytes)", progress, total_bytes, file_size);
            }
        }
        
        data_writer.flush().await?;
        drop(data_writer); // 关闭数据连接
        
        // 读取传输完成响应
        let response = self.read_response().await?;
        let duration = start_time.elapsed();
        
        if response.code == FtpResponseCode::ClosingDataConnection {
            // 计算校验和（如果启用）
            let checksum_result = if let Some(checksum_type) = checksum_type {
                Some(self.calculate_checksum(local_path, checksum_type).await?)
            } else {
                None
            };
            
            Ok(TransferResult {
                bytes_transferred: total_bytes - resume_offset,
                duration,
                speed: (total_bytes - resume_offset) as f64 / duration.as_secs_f64(),
                resumed,
                resume_offset,
                checksum: checksum_result,
                integrity_verified: true, // 上传时假设完整性验证通过
            })
        } else {
            Err(anyhow::anyhow!("Upload failed: {}", response.message))
        }
    }
    
    /// 下载文件（支持断点续传）
    pub async fn download_file(&mut self, remote_filename: &str, local_path: &Path) -> Result<TransferResult> {
        self.download_file_with_resume(remote_filename, local_path, true).await
    }
    
    /// 下载文件（带断点续传选项）
    pub async fn download_file_with_resume(&mut self, remote_filename: &str, local_path: &Path, enable_resume: bool) -> Result<TransferResult> {
        self.download_file_with_resume_and_checksum(remote_filename, local_path, enable_resume, None).await
    }
    
    /// 下载文件（带断点续传和校验和选项）
    pub async fn download_file_with_resume_and_checksum(&mut self, remote_filename: &str, local_path: &Path, enable_resume: bool, checksum_type: Option<ChecksumType>) -> Result<TransferResult> {
        self.ensure_authenticated()?;
        
        // 确保使用被动模式
        self.set_transfer_mode(TransferMode::Passive).await?;
        
        // 检查断点续传
        let mut resume_offset = 0u64;
        let mut resumed = false;
        
        if enable_resume && local_path.exists() {
            // 检查本地文件大小
            let local_size = tokio::fs::metadata(local_path).await?.len();
            
            // 检查远程文件大小
            if let Some(remote_size) = self.get_file_size(remote_filename).await? {
                if local_size < remote_size && local_size > 0 {
                    resume_offset = local_size;
                    resumed = true;
                    info!("Resuming download from offset {} bytes", resume_offset);
                    
                    // 设置REST命令
                    self.set_resume_offset(resume_offset).await?;
                }
            }
        }
        
        // 发送RETR命令
        self.send_command(&format!("RETR {}", remote_filename)).await?;
        let response = self.read_response().await?;
        
        if response.code != FtpResponseCode::DataConnectionAlreadyOpen {
            return Err(anyhow::anyhow!("Failed to start download: {}", response.message));
        }
        
        // 建立数据连接
        let data_stream = self.create_data_connection().await?;
        let mut data_reader = BufReader::new(data_stream);
        
        // 打开或创建本地文件
        let mut file = if resumed {
            tokio::fs::OpenOptions::new()
                .create(true)
                .write(true)
                .append(true)
                .open(local_path)
                .await?
        } else {
            tokio::fs::File::create(local_path).await?
        };
        
        // 传输文件
        let start_time = Instant::now();
        let mut buffer = [0u8; 8192];
        let mut total_bytes = resume_offset;
        
        loop {
            let bytes_read = data_reader.read(&mut buffer).await?;
            if bytes_read == 0 {
                break;
            }
            
            file.write_all(&buffer[..bytes_read]).await?;
            total_bytes += bytes_read as u64;
            
            // 显示进度
            if resumed {
                info!("Resume download progress: {} bytes", total_bytes);
            } else {
                info!("Download progress: {} bytes", total_bytes);
            }
        }
        
        file.flush().await?;
        drop(data_reader); // 关闭数据连接
        
        // 读取传输完成响应
        let response = self.read_response().await?;
        let duration = start_time.elapsed();
        
        if response.code == FtpResponseCode::ClosingDataConnection {
            // 计算校验和（如果启用）
            let checksum_result = if let Some(checksum_type) = checksum_type {
                Some(self.calculate_checksum(local_path, checksum_type).await?)
            } else {
                None
            };
            
            // 验证文件完整性（如果有校验和）
            let integrity_verified = if let Some(ref checksum) = checksum_result {
                self.verify_file_integrity(local_path, checksum).await.unwrap_or(false)
            } else {
                true // 没有校验和时假设完整性验证通过
            };
            
            Ok(TransferResult {
                bytes_transferred: total_bytes - resume_offset,
                duration,
                speed: (total_bytes - resume_offset) as f64 / duration.as_secs_f64(),
                resumed,
                resume_offset,
                checksum: checksum_result,
                integrity_verified,
            })
        } else {
            Err(anyhow::anyhow!("Download failed: {}", response.message))
        }
    }
    
    /// 创建数据连接
    async fn create_data_connection(&self) -> Result<TcpStream> {
        if let Some(session) = &self.session {
            if let (Some(ip), Some(port)) = (&session.passive_ip, &session.passive_port) {
                let addr = format!("{}:{}", ip, port);
                let stream = TcpStream::connect(&addr).await?;
                info!("Data connection established to {}", addr);
                Ok(stream)
            } else {
                Err(anyhow::anyhow!("Passive connection not configured"))
            }
        } else {
            Err(anyhow::anyhow!("No active session"))
        }
    }
    
    /// 确保已认证
    fn ensure_authenticated(&self) -> Result<()> {
        if !self.is_authenticated {
            Err(anyhow::anyhow!("Not authenticated. Please login first."))
        } else {
            Ok(())
        }
    }
    
    /// 计算文件MD5校验和
    pub async fn calculate_md5_checksum(&self, file_path: &Path) -> Result<String> {
        let mut file = tokio::fs::File::open(file_path).await?;
        let mut context = md5::Context::new();
        let mut buffer = [0u8; 8192];
        
        loop {
            let bytes_read = file.read(&mut buffer).await?;
            if bytes_read == 0 {
                break;
            }
            context.consume(&buffer[..bytes_read]);
        }
        
        let digest = context.compute();
        Ok(format!("{:x}", digest))
    }
    
    /// 计算文件SHA256校验和
    pub async fn calculate_sha256_checksum(&self, file_path: &Path) -> Result<String> {
        let mut file = tokio::fs::File::open(file_path).await?;
        let mut hasher = Sha256::new();
        let mut buffer = [0u8; 8192];
        
        loop {
            let bytes_read = file.read(&mut buffer).await?;
            if bytes_read == 0 {
                break;
            }
            hasher.update(&buffer[..bytes_read]);
        }
        
        let result = hasher.finalize();
        Ok(hex::encode(result))
    }
    
    /// 计算文件校验和
    pub async fn calculate_checksum(&self, file_path: &Path, checksum_type: ChecksumType) -> Result<ChecksumResult> {
        let checksum = match checksum_type {
            ChecksumType::Md5 => self.calculate_md5_checksum(file_path).await?,
            ChecksumType::Sha256 => self.calculate_sha256_checksum(file_path).await?,
        };
        
        Ok(ChecksumResult {
            checksum_type,
            checksum,
        })
    }
    
    /// 验证文件完整性
    pub async fn verify_file_integrity(&self, file_path: &Path, expected_checksum: &ChecksumResult) -> Result<bool> {
        let actual_checksum = self.calculate_checksum(file_path, expected_checksum.checksum_type.clone()).await?;
        Ok(actual_checksum.checksum == expected_checksum.checksum)
    }
    
    async fn send_command(&mut self, command: &str) -> Result<()> {
        if let Some(stream) = &mut self.control_stream {
            let command_str = format!("{}\r\n", command);
            debug!("Sending command: {}", command.trim());
            stream.get_mut().write_all(command_str.as_bytes()).await?;
            stream.get_mut().flush().await?;
        }
        Ok(())
    }
    
    async fn read_response(&mut self) -> Result<FtpResponse> {
        if let Some(stream) = &mut self.control_stream {
            let mut line = String::new();
            stream.read_line(&mut line).await?;
            
            let line = line.trim();
            debug!("Received response: {}", line);
            
            if let Some((code_str, message)) = line.split_once(' ') {
                if let Ok(code) = code_str.parse::<u16>() {
                    let response_code = match code {
                        125 => FtpResponseCode::DataConnectionAlreadyOpen,
                        150 => FtpResponseCode::FileStatusOk,
                        200 => FtpResponseCode::CommandOk,
                        215 => FtpResponseCode::SystemType,
                        220 => FtpResponseCode::ServiceReady,
                        221 => FtpResponseCode::ServiceClosing,
                        225 => FtpResponseCode::DataConnectionOpen,
                        226 => FtpResponseCode::ClosingDataConnection,
                        227 => FtpResponseCode::EnteringPassiveMode,
                        230 => FtpResponseCode::UserLoggedIn,
                        250 => FtpResponseCode::RequestedFileActionOk,
                        257 => FtpResponseCode::PathnameCreated,
                        331 => FtpResponseCode::UserNameOk,
                        421 => FtpResponseCode::ServiceNotAvailable,
                        425 => FtpResponseCode::DataConnectionNotOpen,
                        426 => FtpResponseCode::ConnectionClosed,
                        450 => FtpResponseCode::FileActionNotTaken,
                        451 => FtpResponseCode::LocalError,
                        452 => FtpResponseCode::InsufficientStorage,
                        500 => FtpResponseCode::SyntaxError,
                        501 => FtpResponseCode::ParameterSyntaxError,
                        502 => FtpResponseCode::CommandNotImplemented,
                        503 => FtpResponseCode::BadSequence,
                        504 => FtpResponseCode::CommandNotImplementedForParameter,
                        530 => FtpResponseCode::NotLoggedIn,
                        550 => FtpResponseCode::FileNotFound,
                        551 => FtpResponseCode::PageTypeUnknown,
                        552 => FtpResponseCode::ExceededStorageAllocation,
                        553 => FtpResponseCode::FileNameNotAllowed,
                        _ => FtpResponseCode::CommandOk, // 默认
                    };
                    
                    return Ok(FtpResponse::new(response_code, message));
                }
            }
            
            Err(anyhow::anyhow!("Invalid response format: {}", line))
        } else {
            Err(anyhow::anyhow!("Not connected"))
        }
    }
    
    /// 获取并发传输管理器的引用
    pub fn get_concurrent_manager(&self) -> Option<&ConcurrentTransferManager> {
        self.concurrent_manager.as_ref()
    }
    
    /// 启动并发传输管理器
    pub async fn start_concurrent_transfers(&mut self) -> Result<()> {
        if let Some(manager) = &self.concurrent_manager {
            manager.start_queue_processor().await?;
        }
        Ok(())
    }
    
    /// 停止并发传输管理器
    pub async fn stop_concurrent_transfers(&mut self) -> Result<()> {
        if let Some(manager) = &self.concurrent_manager {
            manager.stop_queue_processor().await?;
        }
        Ok(())
    }
    
    
    /// 获取传输统计信息
    pub async fn get_transfer_stats(&self) -> Result<TransferStats> {
        if let Some(manager) = &self.concurrent_manager {
            Ok(manager.get_transfer_stats().await)
        } else {
            Err(anyhow::anyhow!("Concurrent transfer manager not initialized"))
        }
    }
    
    /// 取消任务
    pub async fn cancel_transfer_task(&mut self, task_id: u64) -> Result<()> {
        if let Some(manager) = &self.concurrent_manager {
            manager.cancel_task(task_id).await
        } else {
            Err(anyhow::anyhow!("Concurrent transfer manager not initialized"))
        }
    }
    
    /// 重新设置任务优先级
    pub async fn reprioritize_transfer_task(&mut self, task_id: u64, new_priority: TransferPriority) -> Result<()> {
        if let Some(manager) = &self.concurrent_manager {
            manager.reprioritize_task(task_id, new_priority).await
        } else {
            Err(anyhow::anyhow!("Concurrent transfer manager not initialized"))
        }
    }
    
    /// 获取任务详情
    pub async fn get_transfer_task_details(&self, task_id: u64) -> Result<Option<TransferTaskInfo>> {
        if let Some(manager) = &self.concurrent_manager {
            Ok(manager.get_task_details(task_id).await)
        } else {
            Err(anyhow::anyhow!("Concurrent transfer manager not initialized"))
        }
    }
}

impl ConcurrentTransferManager {
    /// 创建新的并发传输管理器
    pub fn new(max_concurrent: usize) -> Self {
        Self::with_config(max_concurrent, 10, true, false) // 默认批处理大小10，启用预取，禁用压缩
    }
    
    /// 使用配置创建并发传输管理器
    pub fn with_config(max_concurrent: usize, batch_size: usize, prefetch_enabled: bool, compression_enabled: bool) -> Self {
        // 初始化缓冲区池
        let mut buffer_pool = Vec::new();
        for _ in 0..max_concurrent {
            buffer_pool.push(vec![0u8; 65536]); // 64KB缓冲区
        }
        
        Self {
            max_concurrent,
            semaphore: Arc::new(Semaphore::new(max_concurrent)),
            priority_queue: Arc::new(TokioMutex::new(BinaryHeap::new())),
            task_map: Arc::new(TokioMutex::new(std::collections::HashMap::new())),
            running_tasks: Arc::new(TokioMutex::new(Vec::new())),
            completed_tasks: Arc::new(TokioMutex::new(Vec::new())),
            next_task_id: Arc::new(Mutex::new(1)),
            queue_processor_handle: Arc::new(TokioMutex::new(None)),
            buffer_pool: Arc::new(Mutex::new(buffer_pool)),
            batch_size,
            prefetch_enabled,
            compression_enabled,
            transfer_stats: Arc::new(Mutex::new(TransferStats::default())),
        }
    }
    
    /// 添加传输任务
    pub async fn add_task(&self, task: TransferTask, priority: TransferPriority) -> u64 {
        let mut next_id = self.next_task_id.lock().unwrap();
        let task_id = *next_id;
        *next_id += 1;
        
        let now = Instant::now();
        let task_info = TransferTaskInfo {
            id: task_id,
            task: task.clone(),
            priority: priority.clone(),
            status: TransferStatus::Pending,
            progress: 0.0,
            bytes_transferred: 0,
            total_bytes: None,
            speed: 0.0,
            error: None,
            result: None,
            created_at: now,
            started_at: None,
            completed_at: None,
        };
        
        // 添加到任务映射
        let mut task_map = self.task_map.lock().await;
        task_map.insert(task_id, task_info);
        
        // 添加到优先级队列
        let mut priority_queue = self.priority_queue.lock().await;
        priority_queue.push(PriorityQueueItem {
            priority,
            created_at: now,
            task_id,
        });
        
        task_id
    }
    
    /// 添加传输任务（默认优先级）
    pub async fn add_task_default_priority(&self, task: TransferTask) -> u64 {
        self.add_task(task, TransferPriority::Normal).await
    }
    
    /// 启动队列处理器
    pub async fn start_queue_processor(&self) -> Result<()> {
        let mut handle = self.queue_processor_handle.lock().await;
        if handle.is_some() {
            return Ok(()); // 已经在运行
        }
        
        let semaphore = self.semaphore.clone();
        let priority_queue = self.priority_queue.clone();
        let task_map = self.task_map.clone();
        let running_tasks = self.running_tasks.clone();
        let completed_tasks = self.completed_tasks.clone();
        
        let processor_handle = tokio::spawn(async move {
            loop {
                // 等待信号量许可
                let _permit = semaphore.acquire().await.unwrap();
                
                // 从优先级队列获取下一个任务
                let next_task_id = {
                    let mut queue = priority_queue.lock().await;
                    queue.pop().map(|item| item.task_id)
                };
                
                if let Some(task_id) = next_task_id {
                    // 获取任务信息
                    let task_info = {
                        let mut task_map = task_map.lock().await;
                        task_map.get_mut(&task_id).map(|task| {
                            task.status = TransferStatus::Running;
                            task.started_at = Some(Instant::now());
                            task.clone()
                        })
                    };
                    
                    if let Some(task_info) = task_info {
                        // 添加到运行任务列表
                        {
                            let mut running = running_tasks.lock().await;
                            running.push(task_info.clone());
                        }
                        
                        // 执行任务
                        let result = Self::execute_transfer_task(task_info.task.clone()).await;
                        
                        // 更新任务状态
                        {
                            let mut task_map = task_map.lock().await;
                            if let Some(task) = task_map.get_mut(&task_id) {
                                match result {
                                    Ok(transfer_result) => {
                                        task.status = TransferStatus::Completed;
                                        task.result = Some(transfer_result);
                                    }
                                    Err(_) => {
                                        task.status = TransferStatus::Failed;
                                    }
                                }
                                task.completed_at = Some(Instant::now());
                            }
                        }
                        
                        // 移动到完成任务列表
                        {
                            let mut running = running_tasks.lock().await;
                            let mut completed = completed_tasks.lock().await;
                            if let Some(pos) = running.iter().position(|t| t.id == task_id) {
                                let task = running.remove(pos);
                                completed.push(task);
                            }
                        }
                    }
                } else {
                    // 没有任务，等待一段时间
                    tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
                }
            }
        });
        
        *handle = Some(processor_handle);
        Ok(())
    }
    
    /// 停止队列处理器
    pub async fn stop_queue_processor(&self) -> Result<()> {
        let mut handle = self.queue_processor_handle.lock().await;
        if let Some(handle) = handle.take() {
            handle.abort();
        }
        Ok(())
    }
    
    /// 执行传输任务
    async fn execute_transfer_task(task: TransferTask) -> Result<TransferResult> {
        match task {
            TransferTask::Upload { remote_path: _, local_path: _, checksum_type, enable_resume } => {
                // 这里应该调用实际的FTP上传逻辑
                // 为了演示，我们模拟一个上传过程
                tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
                Ok(TransferResult {
                    bytes_transferred: 1024,
                    duration: std::time::Duration::from_millis(100),
                    speed: 10240.0,
                    resumed: enable_resume,
                    resume_offset: if enable_resume { 0 } else { 0 },
                    checksum: checksum_type.map(|ct| ChecksumResult {
                        checksum_type: ct,
                        checksum: "abc123".to_string(),
                    }),
                    integrity_verified: true,
                })
            }
            TransferTask::Download { remote_path: _, local_path: _, checksum_type, enable_resume } => {
                // 这里应该调用实际的FTP下载逻辑
                // 为了演示，我们模拟一个下载过程
                tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
                Ok(TransferResult {
                    bytes_transferred: 2048,
                    duration: std::time::Duration::from_millis(100),
                    speed: 20480.0,
                    resumed: enable_resume,
                    resume_offset: if enable_resume { 0 } else { 0 },
                    checksum: checksum_type.map(|ct| ChecksumResult {
                        checksum_type: ct,
                        checksum: "def456".to_string(),
                    }),
                    integrity_verified: true,
                })
            }
        }
    }
    
    /// 获取待处理任务
    pub async fn get_pending_tasks(&self) -> Vec<TransferTaskInfo> {
        let tasks = self.task_map.lock().await;
        tasks.values()
            .filter(|task| task.status == TransferStatus::Pending)
            .cloned()
            .collect()
    }
    
    /// 获取运行中任务
    pub async fn get_running_tasks(&self) -> Vec<TransferTaskInfo> {
        let running_tasks = self.running_tasks.lock().await;
        running_tasks.clone()
    }
    
    /// 获取已完成任务
    pub async fn get_completed_tasks(&self) -> Vec<TransferTaskInfo> {
        let completed_tasks = self.completed_tasks.lock().await;
        completed_tasks.clone()
    }
    
    /// 获取所有任务状态
    pub async fn get_all_tasks(&self) -> Vec<TransferTaskInfo> {
        let mut all_tasks = Vec::new();
        
        // 添加所有任务（从task_map获取）
        let task_map = self.task_map.lock().await;
        all_tasks.extend(task_map.values().cloned());
        
        all_tasks
    }
    
    /// 更新任务状态
    pub async fn update_task_status(&self, task_id: u64, status: TransferStatus) -> Result<()> {
        let mut task_map = self.task_map.lock().await;
        if let Some(task) = task_map.get_mut(&task_id) {
            task.status = status.clone();
            
            match status {
                TransferStatus::Running => {
                    task.started_at = Some(Instant::now());
                    let mut running_tasks = self.running_tasks.lock().await;
                    running_tasks.push(task.clone());
                }
                TransferStatus::Completed | TransferStatus::Failed | TransferStatus::Cancelled => {
                    task.completed_at = Some(Instant::now());
                    let mut completed_tasks = self.completed_tasks.lock().await;
                    completed_tasks.push(task.clone());
                }
                _ => {}
            }
        }
        
        Ok(())
    }
    
    /// 更新任务进度
    pub async fn update_task_progress(&self, task_id: u64, progress: f64, bytes_transferred: u64, speed: f64) -> Result<()> {
        let mut task_map = self.task_map.lock().await;
        if let Some(task) = task_map.get_mut(&task_id) {
            task.progress = progress;
            task.bytes_transferred = bytes_transferred;
            task.speed = speed;
        }
        
        Ok(())
    }
    
    /// 设置任务错误
    pub async fn set_task_error(&self, task_id: u64, error: String) -> Result<()> {
        self.update_task_status(task_id, TransferStatus::Failed).await?;
        
        let mut task_map = self.task_map.lock().await;
        if let Some(task) = task_map.get_mut(&task_id) {
            task.error = Some(error);
        }
        
        Ok(())
    }
    
    /// 设置任务结果
    pub async fn set_task_result(&self, task_id: u64, result: TransferResult) -> Result<()> {
        self.update_task_status(task_id, TransferStatus::Completed).await?;
        
        let mut task_map = self.task_map.lock().await;
        if let Some(task) = task_map.get_mut(&task_id) {
            task.result = Some(result);
        }
        
        Ok(())
    }
    
    /// 取消任务
    pub async fn cancel_task(&self, task_id: u64) -> Result<()> {
        self.update_task_status(task_id, TransferStatus::Cancelled).await
    }
    
    /// 清空已完成任务
    pub async fn clear_completed_tasks(&self) {
        let mut completed_tasks = self.completed_tasks.lock().await;
        completed_tasks.clear();
    }
    
    /// 获取最大并发数
    pub fn max_concurrent(&self) -> usize {
        self.max_concurrent
    }
    
    /// 设置最大并发数
    pub fn set_max_concurrent(&mut self, max_concurrent: usize) {
        self.max_concurrent = max_concurrent;
        self.semaphore = Arc::new(Semaphore::new(max_concurrent));
    }
    
    /// 获取队列长度
    pub async fn get_queue_length(&self) -> usize {
        let priority_queue = self.priority_queue.lock().await;
        priority_queue.len()
    }
    
    /// 获取下一个要执行的任务
    pub async fn get_next_task(&self) -> Option<u64> {
        let mut priority_queue = self.priority_queue.lock().await;
        priority_queue.pop().map(|item| item.task_id)
    }
    
    /// 重新排列任务优先级
    pub async fn reprioritize_task(&self, task_id: u64, new_priority: TransferPriority) -> Result<()> {
        let mut task_map = self.task_map.lock().await;
        if let Some(task) = task_map.get_mut(&task_id) {
            if task.status == TransferStatus::Pending {
                task.priority = new_priority.clone();
                
                // 重新添加到优先级队列
                let mut priority_queue = self.priority_queue.lock().await;
                priority_queue.push(PriorityQueueItem {
                    priority: new_priority,
                    created_at: task.created_at,
                    task_id,
                });
            }
        }
        
        Ok(())
    }
    
    /// 暂停队列处理
    pub async fn pause_queue(&self) -> Result<()> {
        let mut handle = self.queue_processor_handle.lock().await;
        if let Some(h) = handle.take() {
            h.abort();
        }
        Ok(())
    }
    
    /// 恢复队列处理
    pub async fn resume_queue(&self) -> Result<()> {
        // 这里需要实现队列处理器的启动逻辑
        // 暂时留空，后续实现
        Ok(())
    }
    
    /// 清空队列
    pub async fn clear_queue(&self) {
        let mut priority_queue = self.priority_queue.lock().await;
        priority_queue.clear();
        
        let mut task_map = self.task_map.lock().await;
        task_map.retain(|_, task| task.status != TransferStatus::Pending);
    }
    
    /// 获取队列统计信息
    pub async fn get_queue_stats(&self) -> QueueStats {
        let task_map = self.task_map.lock().await;
        let priority_queue = self.priority_queue.lock().await;
        let running_tasks = self.running_tasks.lock().await;
        let completed_tasks = self.completed_tasks.lock().await;
        
        let mut stats = QueueStats {
            total_tasks: task_map.len(),
            pending_tasks: 0,
            running_tasks: running_tasks.len(),
            completed_tasks: completed_tasks.len(),
            failed_tasks: 0,
            cancelled_tasks: 0,
            queue_length: priority_queue.len(),
        };
        
        for task in task_map.values() {
            match task.status {
                TransferStatus::Pending => stats.pending_tasks += 1,
                TransferStatus::Failed => stats.failed_tasks += 1,
                TransferStatus::Cancelled => stats.cancelled_tasks += 1,
                _ => {} // Running和Completed已经在上面计算了
            }
        }
        
        stats
    }
    
    /// 获取传输统计信息
    pub async fn get_transfer_stats(&self) -> TransferStats {
        let task_map = self.task_map.lock().await;
        let running_tasks = self.running_tasks.lock().await;
        let completed_tasks = self.completed_tasks.lock().await;
        
        let mut total_bytes = 0u64;
        let mut total_duration = std::time::Duration::new(0, 0);
        let mut completed_count = 0;
        
        // 计算已完成任务的统计
        for task in completed_tasks.iter() {
            if let Some(result) = &task.result {
                total_bytes += result.bytes_transferred;
                total_duration += result.duration;
                completed_count += 1;
            }
        }
        
        // 计算当前运行任务的统计
        for task in running_tasks.iter() {
            total_bytes += task.bytes_transferred;
        }
        
        let avg_speed = if total_duration.as_millis() > 0 {
            (total_bytes as f64) / (total_duration.as_millis() as f64) * 1000.0
        } else {
            0.0
        };
        
        TransferStats {
            total_transfers: task_map.len() as u64,
            successful_transfers: completed_count as u64,
            failed_transfers: task_map.values().filter(|t| t.status == TransferStatus::Failed).count() as u64,
            total_bytes,
            average_speed: avg_speed,
            peak_speed: 0.0, // TODO: 计算峰值速度
            total_duration,
        }
    }
    
    /// 获取任务详情
    pub async fn get_task_details(&self, task_id: u64) -> Option<TransferTaskInfo> {
        let task_map = self.task_map.lock().await;
        task_map.get(&task_id).cloned()
    }
}


/// 队列统计信息
#[derive(Debug, Clone)]
pub struct QueueStats {
    pub total_tasks: usize,
    pub pending_tasks: usize,
    pub running_tasks: usize,
    pub completed_tasks: usize,
    pub failed_tasks: usize,
    pub cancelled_tasks: usize,
    pub queue_length: usize,
}

impl ConcurrentTransferManager {
    /// 获取传输统计（优化版本）
    pub fn get_optimized_transfer_stats(&self) -> TransferStats {
        self.transfer_stats.lock().unwrap().clone()
    }
    
    /// 更新传输统计
    fn update_transfer_stats(&self, result: &TransferResult) {
        let mut stats = self.transfer_stats.lock().unwrap();
        stats.total_transfers += 1;
        stats.successful_transfers += 1;
        stats.total_bytes += result.bytes_transferred;
        stats.total_duration += result.duration;
        
        if result.speed > stats.peak_speed {
            stats.peak_speed = result.speed;
        }
        
        if stats.successful_transfers > 0 {
            stats.average_speed = stats.total_bytes as f64 / stats.total_duration.as_secs_f64();
        }
    }
    
    /// 获取缓冲区（从池中获取）
    pub fn get_buffer(&self) -> Option<Vec<u8>> {
        self.buffer_pool.lock().unwrap().pop()
    }
    
    /// 归还缓冲区到池中
    pub fn return_buffer(&self, mut buffer: Vec<u8>) {
        buffer.clear(); // 清空缓冲区内容
        self.buffer_pool.lock().unwrap().push(buffer);
    }
    
    /// 批量处理传输任务
    pub async fn process_batch(&self) -> Result<()> {
        let mut tasks_to_process = Vec::new();
        
        // 获取一批任务
        {
            let mut queue = self.priority_queue.lock().await;
            for _ in 0..self.batch_size {
                if let Some(item) = queue.pop() {
                    tasks_to_process.push(item);
                } else {
                    break;
                }
            }
        }
        
        if tasks_to_process.is_empty() {
            return Ok(());
        }
        
        // 并发处理这批任务
        let mut handles = Vec::new();
        
        for _item in tasks_to_process {
            let manager = self.clone();
            let handle = tokio::spawn(async move {
                let _permit = manager.semaphore.acquire().await.unwrap();
                // 执行任务逻辑
                // manager.execute_task(item.task_id).await;
            });
            handles.push(handle);
        }
        
        // 等待所有任务完成
        for handle in handles {
            let _ = handle.await;
        }
        
        Ok(())
    }
    
    /// 预取文件数据
    pub async fn prefetch_file(&self, file_path: &str) -> Result<Vec<u8>> {
        if !self.prefetch_enabled {
            return Ok(Vec::new());
        }
        
        // 读取文件的前几KB用于预取
        let mut file = tokio::fs::File::open(file_path).await?;
        let mut buffer = vec![0u8; 8192]; // 8KB预取
        let bytes_read = file.read(&mut buffer).await?;
        buffer.truncate(bytes_read);
        
        Ok(buffer)
    }
    
    /// 压缩数据
    pub fn compress_data(&self, data: &[u8]) -> Result<Vec<u8>> {
        if !self.compression_enabled {
            return Ok(data.to_vec());
        }
        
        // 使用简单的压缩（实际项目中可以使用更高效的压缩算法）
        use std::io::Write;
        let mut encoder = flate2::write::GzEncoder::new(Vec::new(), flate2::Compression::default());
        encoder.write_all(data)?;
        Ok(encoder.finish()?)
    }
    
    /// 解压数据
    pub fn decompress_data(&self, data: &[u8]) -> Result<Vec<u8>> {
        if !self.compression_enabled {
            return Ok(data.to_vec());
        }
        
        use std::io::Read;
        let mut decoder = flate2::read::GzDecoder::new(data);
        let mut buffer = Vec::new();
        decoder.read_to_end(&mut buffer)?;
        Ok(buffer)
    }
}
